package com.cssl.app.dws;

import com.cssl.bean.ProvinceStats;
import com.cssl.utils.ClickHouseUtil;
import com.cssl.utils.CommonUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Properties;

/**
 * @Author: chen
 * @Date: 2021/11/18 17:31
 * @Desc: 地区主题宽表
 */
public class ProvinceStatsSqlApp {
    public static void main(String[] args) throws Exception {
        //1. 获取执行环境
        Properties properties = CommonUtils.getProperties();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //2. 使用DDL创建表 提取时间戳生成watermark
        String sourceTopic = properties.getProperty("kafka.topic.db.dwm.orderWide");
        String groupId = properties.getProperty("kafka.consumer.group");
        tableEnv.executeSql("CREATE TABLE order_wide ( " +
                "  province_id BIGINT, " +
                "  province_name STRING, " +
                "  province_area_code STRING, " +
                "  province_iso_code STRING, " +
                "  province_3166_2_code STRING, " +
                "  order_id BIGINT, " +
                "  split_total_amount Decimal, " +
                "  create_time STRING, " +
                "  rt as TO_TIMESTAMP(create_time), " +
                "  WATERMARK FOR rt as rt - INTERVAL '1' SECOND ) WITH ( " +
                CommonUtils.getKafkaDDL(sourceTopic, groupId) + ")");

        //3. 查询数据 分组,开窗,聚合
        Table table = tableEnv.sqlQuery("select " +
                "    DATE_FORMAT(TUMBLE_START(rt,INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') stt, " +
                "    DATE_FORMAT(TUMBLE_END(rt,INTERVAL '10' SECOND),'yyyy-MM-dd HH:mm:ss') edt, " +
                "    province_id, " +
                "    province_name, " +
                "    province_area_code, " +
                "    province_iso_code, " +
                "    province_3166_2_code, " +
                "    sum(split_total_amount) order_amount, " +
                "    count(distinct order_id) order_count, " +
                "    UNIX_TIMESTAMP()*1000 as ts " +
                "from " +
                "    order_wide " +
                "group by " +
                "    province_id, " +
                "    province_name, " +
                "    province_area_code, " +
                "    province_iso_code, " +
                "    province_3166_2_code, " +
                "    TUMBLE(rt,INTERVAL '10' SECOND)");

        //4. 将动态表转换为流
        DataStream<ProvinceStats> provinceStatsDS = tableEnv.toAppendStream(table, ProvinceStats.class);

        //5. 打印数据并写入 clickhouse
        provinceStatsDS.print();
        provinceStatsDS.addSink(ClickHouseUtil.getSink("insert into table province_stats_2021 values(?,?,?,?,?,?,?,?,?,?)"));

        //6. 启动任务
        env.execute("ProvinceStatsSqlApp");
    }
}
